package cn.lecosa.spark.hbase

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.mapred.JobConf
import org.apache.spark._
import org.apache.spark.SparkConf
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.hive.HiveContext

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes

//http://www.cnblogs.com/smartcuning/p/4595239.html
object Hive2Hbase {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getName}");
    conf.setMaster("local[2]");
    val sc = new SparkContext(conf);
    val sqlContext = new SQLContext(sc);
    val hiveContext = new HiveContext(sc);
    import sqlContext.implicits._
    val df = hiveContext.sql("SELECT * from  lecosa.h_ticket")

    //创建jobConf
    val hbaseconf = HBaseConfiguration.create(sc.hadoopConfiguration)
    val jobConf = new JobConf(hbaseconf)
    jobConf.setOutputFormat(classOf[TableOutputFormat])
    jobConf.set(TableOutputFormat.OUTPUT_TABLE, "tab_x_1")

    //创建hiveContext
    hiveContext.setConf("spark.sql.shuffle.partitions", "3")
    //保存到hbase
    df.rdd.map(row => {
      val c1 = row(0).asInstanceOf[String]
      val c2 = row(1).asInstanceOf[String]
      val c3 = row(2).asInstanceOf[String]
      val c4 = row(3).asInstanceOf[String]
      val c5 = row(4).asInstanceOf[String]
      val c6 = row(5).asInstanceOf[String]
      val c7 = row(6).asInstanceOf[String]

      val p = new Put(Bytes.toBytes(c4))
      p.add(Bytes.toBytes("cf1"), Bytes.toBytes("c1"), Bytes.toBytes(c1))
      p.add(Bytes.toBytes("cf1"), Bytes.toBytes("c2"), Bytes.toBytes(c2))
      p.add(Bytes.toBytes("cf1"), Bytes.toBytes("c3"), Bytes.toBytes(c3))
      p.add(Bytes.toBytes("cf1"), Bytes.toBytes("c4"), Bytes.toBytes(c4))
      p.add(Bytes.toBytes("cf1"), Bytes.toBytes("c5"), Bytes.toBytes(c5))
      p.add(Bytes.toBytes("cf1"), Bytes.toBytes("c6"), Bytes.toBytes(c6))
      p.add(Bytes.toBytes("cf1"), Bytes.toBytes("c7"), Bytes.toBytes(c7))
      (new ImmutableBytesWritable, p)
    }).saveAsHadoopDataset(jobConf)

    df.show();
    df.schema;
    sc.stop()
  }

}